package cn.seecoder.ai.utils;


import cn.seecoder.ai.exception.AIInternalException;

import lombok.Data;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.web.multipart.MultipartFile;
import javax.annotation.PostConstruct;
import javax.servlet.ServletOutputStream;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.InputStream;
import java.net.URLEncoder;
import java.util.Date;

/**
 * @author fanyanpeng
 * @date 2023/4/7 21:28
 */
@Slf4j
@Data
@Component
public class HdfsHelper {

    @Value("${hadoop.hdfsUrl}")
    private String hdfsUrl;

    @Value("${hadoop.hdfsUrl}user-space/${user.name}/")
    private String hdfsUserSpaceUrl;  //get hdfs Address

    // 操作hdfs的文件系统接口
    private FileSystem hdfs;

    @PostConstruct
    private void initHdfsHelper(){
        useUserSpace();
    }

    public String getUriPrediciton(Integer modelId,Integer fileInfoId){
        long timeStamp = new Date().getTime();
        return this.hdfsUserSpaceUrl + "prediction/model-"+modelId+"/"+fileInfoId+"_"+timeStamp+".csv";
    }

    public String getUriUploadFile(Integer userId,String fileName){
        return this.hdfsUserSpaceUrl + "upload/user-"+userId+"/"+fileName;
    }

    public String getUriConvertToParquetFile(Integer userId,String fileName){
        return this.hdfsUserSpaceUrl + "parquet/user-"+userId+"/"+fileName;
    }

    public String getUriPipelineModel(Integer userId,Integer modelId){
        return this.hdfsUserSpaceUrl + "model/pipeline-model/user-"+userId+"/"+modelId;
    }

    public String getUriFeatureStringIndexerPipeline(Integer modelId){
        return this.hdfsUserSpaceUrl + "model/feature-string-indexer-pipeline/"+modelId;
    }

    public String getUriLabelStringIndexer(Integer modelId){
        return this.hdfsUserSpaceUrl + "model/label-string-indexer/"+modelId;
    }

    public String getUriLibSvm(Integer modelId,Integer fileInfoId){
        return this.hdfsUserSpaceUrl + "model/"+modelId+"/libsvm/"+fileInfoId;
    }




    public String getUploadFilePathString(String fileName){
        return this.hdfsUserSpaceUrl + "upload/" + fileName;
    }

    public String getUploadFilePathByRelativePath(String relativePath){
        return this.hdfsUserSpaceUrl + relativePath;
    }

    public String saveMultipartFile(MultipartFile multipartFile,String fileUri){

        try {
            InputStream in = multipartFile.getInputStream();
            FSDataOutputStream out =  hdfs.create(new Path(fileUri),true);
            IOUtils.copyBytes(in,out,1024,true);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return fileUri;
    }


    // 找到文件名
    private String getFileNameFromFileUri(String fileUri){
        int indexOfLast = fileUri.lastIndexOf('/');
        String fileName = fileUri.substring(indexOfLast+1);
        assert !fileName.isEmpty();
        return fileName;

    }

    @SneakyThrows
    public  void downloadSingleFileGeneratedBySparkFromFileFolderUri(String fileFolderUri, String format, HttpServletResponse httpServletResponse){
        // read sub files
        FileStatus[] fileStatuses = hdfs.listStatus(new Path(fileFolderUri));
        FileStatus selectedFile = null;
        for(FileStatus fileStatus : fileStatuses){
            if(fileStatus.getPath().getName().endsWith(format)){
                selectedFile = fileStatus;
                break;
            }
        }
        if(selectedFile==null){
            throw new AIInternalException("没有匹配的文件");
        }

        String fileUri = selectedFile.getPath().toString();
        downloadFileByFileUri(fileUri,httpServletResponse);
        // select csv file
        // get csv fileUri
    }

    /**
     * 给定文件uri，下载文件：已验证：单个文件下载-OK；未验证：spark自动保存文件-FAIL
     * 必须是单个文件，不能保存为文件夹。
     * @author   fanyanpeng
     * @date 2023/4/8 1:53
     * @param fileUri 文件的uri
     * @param httpServletResponse 通过response将数据传递回前端，此处的数据为文件
     * @return 文件通过response返回
     */
    @SneakyThrows
    public void downloadFileByFileUri(String fileUri, HttpServletResponse httpServletResponse) {
        log.info("downloading file: "+fileUri);
        //生成的uri
        String fileName = getFileNameFromFileUri(fileUri);
        fileUri=getRealFileUri(fileUri);
        //在spark中实际保存位置
        String fileNameGeneratedBySpark = getFileNameFromFileUri(fileUri);
        if(!fileNameGeneratedBySpark.contains(".")){
            fileName+=".csv";
        }
        else {
            String postfix = fileNameGeneratedBySpark.substring(fileNameGeneratedBySpark.lastIndexOf("."));
            if(!fileName.endsWith(postfix)){
                fileName+=postfix;
            }
        }
        String nameWithOutPostfix = fileName.substring(0,fileName.lastIndexOf("."));
        String postfix = fileName.substring(fileName.lastIndexOf("."));

        FSDataInputStream in = hdfs.open(new Path(fileUri));
        httpServletResponse.reset();
        httpServletResponse.setContentType("application/octet-stream;charset=UTF-8");
        httpServletResponse.addHeader("Content-Disposition","attachment; filename="+ URLEncoder.encode(nameWithOutPostfix,"UTF-8")+postfix);
        ServletOutputStream outputStream = httpServletResponse.getOutputStream();
        IOUtils.copyBytes(in,outputStream,1024);
        in.close();
    }

    /**
     * 返回真正的文件位置
     * @author   fanyanpeng
     * @date 2023/4/22 1:55
     * @param fileUri
     * @return java.lang.String
     */
    private String getRealFileUri(String fileUri) {

        try{
            FileStatus[] fileStatuses = hdfs.listStatus(new Path(fileUri));
            FileStatus selectedFile = null;
            for(FileStatus fileStatus : fileStatuses){
                if(fileStatus.getPath().getName().contains(".")){
                    selectedFile = fileStatus;
                    break;
                }
            }
            if(selectedFile!=null){
                fileUri = selectedFile.getPath().toString();
            }
        }catch (IOException ioException){
            log.info("本路径下没有需要下载的文件，使用本文件");
        }
        return fileUri;
    }
    // Constructor(构造方法) -> @Autowired(依赖注入) -> @PostConstruct(注释的方法),


    private void useUserSpace() {
        ensureExistenceUserSpace();
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", hdfsUserSpaceUrl);

        try {
            hdfs = FileSystem.get(conf);
            log.info("当前主机下的hdfs目录是：" + hdfsUserSpaceUrl);
        }catch (IOException ioException){
            ioException.printStackTrace();
            throw new HadoopIllegalArgumentException("使用user-space位置失败");
        }
    }

    //保证hdfs存储位置可以访问
    private void ensureExistenceUserSpace(){
        Configuration conf = new Configuration();
        //先根据主机ip设置配置
        conf.set("fs.defaultFS", hdfsUrl);
        try {
            hdfs = FileSystem.get(conf);
            hdfs.mkdirs(new Path(hdfsUserSpaceUrl));
        }catch (IOException ioException){

            throw new HadoopIllegalArgumentException("确保存在user-space位置失败");
        }
    }



}
